package com.guchenbo.spark.core

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter

import com.guchenbo.spark.core.MyIterator.time

import scala.util.Random

/**
 * @author guchenbo
 * @date 2022/4/16
 */
object IterSeq {

  def defaultIter(): Unit = {
    val list = List("hello java", "hello scala", "hello spark", "hello flink")

    val iter1: Iterator[String] = list.toIterator

    val func1: Iterator[String] => Iterator[Array[String]] = (iter: Iterator[String]) => iter.map(_.split(" "))
    val iter2: Iterator[Array[String]] = func1(iter1)

    val func2: Iterator[Array[String]] => Iterator[String] = (iter: Iterator[Array[String]]) => iter.map(_ (1))
    val iter3: Iterator[String] = func2(iter2)

    val li: List[String] = iter3.toList
    println(li)
  }

  def customIter(): Unit = {
    val list = List("hello java", "hello scala", "hello spark", "hello flink")
    val iter: Iterator[String] = list.toIterator

    //     rdd1，连接数据
    val iter1: MyIterator[String] = MyIterator(iter)
    //     rdd2，split每个元素，然后返回新的iter
    val func2: Iterator[String] => Iterator[Array[String]] = (iter: Iterator[String]) => iter.map((str: String) => {
      println("split")
      str.split(" ")
    })
    val iter2: Iterator[Array[String]] = func2(iter1)

    //     rdd3，取某个元素，返回新的iter
    val func3: Iterator[Array[String]] => Iterator[String] = (iter: Iterator[Array[String]]) => iter.map(list => {
      println("get 1")
      list(1)
    })
    val iter3: Iterator[String] = func3(iter2)

    //    collect，这里才会真正触发迭代，在迭代的时候执行上面定义的逻辑
    val li: List[String] = iter3.toList
    println(li)
  }

  def main(args: Array[String]): Unit = {
//    defaultIter()

    customIter()
  }

}

class MyIterator[T](iter: Iterator[T]) extends Iterator[T] {
  override def hasNext: Boolean = iter.hasNext

  override def next(): T = {
    println("my next at", time(),Random.nextInt(10))
    iter.next()
  }
}

object MyIterator {

  def apply[T](iter: Iterator[T]): MyIterator[T] = new MyIterator(iter)

  def time(): String = {
//    System.currentTimeMillis().toString
    LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss,SSS"))
  }
}

